package com.tg.rxjava;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.util.Random;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * rxJava的异步逻辑
 *
 * - 不同Obserable之间串行
 * - 不同Subscriber之间并行
 * - 同类别Obserable之间并行
 *
 * --------------------
 * output：
 * main-start subscribe
 * RxIoScheduler-2-sender1
 * main-end
 * RxIoScheduler-3-sender1
 * RxIoScheduler-2-sender2
 * RxIoScheduler-3-sender2
 * RxIoScheduler-3-subscribe2
 * RxIoScheduler-2-subscribe1
 */
public class RxJavaConcurrent {

    public void execute() throws Exception {

        Observable<Integer> sender2 = Observable.defer(
                new Func0<Observable<Integer>>() {
                    public Observable<Integer> call() {
                        log("sender2");
                        return Observable.just(10);
                    }
                }

        );

        Observable<Integer> sender1 = Observable.defer(
                new Func0<Observable<Integer>>() {
                    public Observable<Integer> call() {
                        log("sender1");
                        return sender2;
                    }
                }

        ).subscribeOn(Schedulers.io());

        log("start subscribe");

        sender1.subscribe(t -> {
            log("subscribe1");
        });

        sender1.subscribe(t -> {
            log("subscribe2");
        });


        log("end");
    }

    public static void main(String[] args) throws Exception {

        RxJavaConcurrent command = new RxJavaConcurrent();
        command.execute();

        TimeUnit.SECONDS.sleep(10);
    }

    private void log(String msg) {
        System.out.println(Thread.currentThread().getName() + "-" + msg);
    }
}
